package com.lzx.spark.sanbao.tcsl.app;

import org.apache.spark.SparkConf;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructType;

import static org.apache.spark.sql.functions.*;

/**
 * @author wangzhen
 * @version 1.0
 * @date 2024/8/24 下午4:38
 * @description: TODO
 */
public class ods_cater_bill_detail_query_billdetail_inc_transfer {
    public static void main(String[] args) {

        System.setProperty("HADOOP_USER_NAME", "hadoop");
// 1 创建spark配置
        SparkConf conf = new SparkConf().setAppName("SparkSql");

        SparkSession spark = SparkSession.builder().enableHiveSupport().config(conf).getOrCreate();

//String yesterday = LocalDate.now().minusDays(1).format(DateTimeFormatter.ofPattern("yyyyMMdd"));

        String yesterday = "20240822";


        StructType shopDataSchema = new StructType()
                .add("bs_id", DataTypes.StringType)
                .add("bs_code", DataTypes.StringType)
                .add("ts_code", DataTypes.StringType)
                .add("owo_open_id", DataTypes.StringType)
                .add("owo_union_id", DataTypes.StringType)// 数组
                .add("btpw_user_account ", DataTypes.StringType)// 数组
                .add("area_code", DataTypes.StringType)
                .add("area_id", DataTypes.StringType)
                .add("area_name", DataTypes.StringType)
                .add("point_code", DataTypes.StringType)
                .add("point_name", DataTypes.StringType)
                .add("people_qty", DataTypes.StringType)
                .add("open_time", DataTypes.StringType)
                .add("settle_time", DataTypes.StringType)
                .add("settleman_id", DataTypes.StringType)
                .add("settleman_name", DataTypes.StringType)
                .add("settle_biz_date", DataTypes.StringType)
                .add("settle_biz_month", DataTypes.StringType)
                .add("settle_shift_id", DataTypes.StringType)
                .add("state", DataTypes.StringType)
                .add("waiter_code", DataTypes.StringType)
                .add("waiter_name", DataTypes.StringType)
                .add("salesman_code", DataTypes.StringType)
                .add("salesman_name", DataTypes.StringType)
                .add("item_orig_money", DataTypes.StringType)
                .add("item_income_total", DataTypes.StringType)
                .add("orig_server_fee", DataTypes.StringType)
                .add("service_fee_income_money", DataTypes.StringType)
                .add("service_fee_not_income_money", DataTypes.StringType)
                .add("service_fee_last_total", DataTypes.StringType)
                .add("zdxf_income_money", DataTypes.StringType)
                .add("orig_zdxfbq", DataTypes.StringType)
                .add("orig_total", DataTypes.StringType)
                .add("disc_total", DataTypes.StringType)
                .add("disc_money", DataTypes.StringType)
                .add("quota_money", DataTypes.StringType)
                .add("present_money", DataTypes.StringType)
                .add("member_money", DataTypes.StringType)
                .add("promote_money", DataTypes.StringType)
                .add("wipe_money", DataTypes.StringType)
                .add("last_total", DataTypes.StringType)
                .add("income_total", DataTypes.StringType)
                .add("not_income_money", DataTypes.StringType)
                .add("sale_type_id", DataTypes.StringType)
                .add("sale_type_name", DataTypes.StringType)
                .add("order_origin_id", DataTypes.StringType)
                .add("order_origin_name", DataTypes.StringType)
                .add("dinner_type_id", DataTypes.StringType)
                .add("dinner_type_name", DataTypes.StringType)
                .add("settle_state", DataTypes.StringType)
                .add("is_give_invoice ", DataTypes.StringType)
                .add("invoice_no", DataTypes.StringType)
                .add("invoice_code", DataTypes.StringType)
                .add("invoice_remark", DataTypes.StringType)
                .add("order_type ", DataTypes.StringType)
                .add("de_from", DataTypes.StringType)
                .add("member_card_no ", DataTypes.StringType)
                .add("member_id", DataTypes.StringType)
                .add("member_name ", DataTypes.StringType)
                .add("member_mobile", DataTypes.StringType)//数组
                .add("card_kind_name ", DataTypes.StringType)
                .add("crm_ts_code", DataTypes.StringType)
                .add("is_continued_bill", DataTypes.StringType)
                .add("is_continued_bill_name", DataTypes.StringType)
                .add("tax_money", DataTypes.StringType)
                .add("order_code", DataTypes.StringType)
                .add("center_code", DataTypes.StringType)
                .add("center_name", DataTypes.StringType)
                .add("brand_code ", DataTypes.StringType)
                .add("brand_name", DataTypes.StringType)
                .add("shop_id", DataTypes.StringType)
                .add("shop_code", DataTypes.StringType)
                .add("shop_name", DataTypes.StringType)
                .add("shop_out_code", DataTypes.StringType)
                .add("delflg", DataTypes.StringType);


// 查询获取 shopid 和 shopdata
        Dataset<Row> showResult = spark.sql("SELECT shopid, shopdata FROM sanbao.ods_cater_bill_detail_query_billdetail_inc_transfer WHERE dt = '" + yesterday + "'");

// 去除头尾的 [] 符号并解析 JSON 数据
        Dataset<Row> parsedData = showResult
                .withColumn("cleanedShopData", expr("substring(shopdata, 2, length(shopdata) - 2)"))
                .withColumn("parsedShopData", from_json(col("cleanedShopData"), DataTypes.createArrayType(shopDataSchema)))
                .withColumn("parsedShopData", explode(col("parsedShopData")))
                .select(col("shopid"), col("parsedShopData.*"));

        parsedData.show(false);


        parsedData = parsedData.withColumn("dt", lit(yesterday));

        parsedData.write()
                .mode("overwrite") // 使用 append 模式插入数据
                .insertInto("sanbao.dwd_cater_bill_detail_query_billdetail_inc");

//itemListParsed.show(false);

        spark.stop();


    }
}


